package com.bw.gmall.realtime.dim.Function;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.Util.HbaseUtil;
import com.bw.gmall.realtime.common.bean.TableProcessDim;

import com.bw.gmall.realtime.common.constant.Constant;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;


public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeConnection(connection);

    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {

        // 主流维度数据，并且已经去除没用的字段
        JSONObject f0 = value.f0;

        // 对应的配置表数据
        TableProcessDim f1 = value.f1;

        // 要往Hbase写入的数据

        JSONObject data = f0.getJSONObject("data");

        // 取出操作类型
        String type = f0.getString("type");

        // 往Hbase哪张表
        String sinkTable = f1.getSinkTable();
        String sinkRowKey = f1.getSinkRowKey();

        // 取出对应的维度数据主键的值
        String sinkRowkeyValue = data.getString(sinkRowKey);

//        取出列簇
        String sinkFamily = f1.getSinkFamily();
        if ("delete".equals(type)) {

//            从hbase删除这条数据

            HbaseUtil.deleteDatas(connection, Constant.HBASE_NAMESPACE,sinkTable,sinkRowkeyValue);
        }else {
            HbaseUtil.putDatas(connection,Constant.HBASE_NAMESPACE,sinkTable,sinkRowkeyValue,sinkFamily,data);
        }

    }
}
